//
// Created by suzhen on 16-12-9.
//

#include "AsyncEventGenerator.h"
#include "AbstractAsyncStream.h"
#include "Utils.h"

USING_ORION_V2

//protected
AsyncEventGenerator::~AsyncEventGenerator() {
    CHECK(openInHandle_ == 0) << "server 仍然被uv对象持有,发生内存泄露";
    delete[] readBuffer_;
    free(generatorCoreLoop_);
}

AsyncEventGenerator::AsyncEventGenerator(size_t readBufSz)
        : generatorCoreLoop_((uv_loop_t *) malloc(sizeof(uv_loop_t))),
          threadId_(),
          readBuffer_(nullptr),
          readBufferSize_(readBufSz),
          openInHandle_(0),
          closePromise_(),
          isClosing_(false),
          internalLock_(),
          acceptQueue_() {
    uv_loop_init(generatorCoreLoop_);
    readBuffer_ = new RawDataType[readBufferSize_];
    //初始化异步stop上下文
    uv_async_init(generatorCoreLoop_, &closeAsyncCtx_, &AsyncEventGenerator::GeneratorAsyncStopCb);
    closeAsyncCtx_.data = this;
    ++openInHandle_;
    //初始化异步accept上下文
    uv_async_init(generatorCoreLoop_, &acceptAsyncCtx_, &AsyncEventGenerator::AsyncStreamAcceptCallBack);
    acceptAsyncCtx_.data = this;
    ++openInHandle_;
}

/**
 * 将 Socket, 压入队列后,在 loop 线程中进行accept
 * @param s
 */
void AsyncEventGenerator::submit(AbstractAsyncStream *s) {
    //加锁后,将Socket压入队列
    {
        //todo mutex 可能会导致性能问题
        std::lock_guard<std::recursive_mutex> guard(internalLock_);
        acceptQueue_.push_back(s);
    }
    //
    if (threadId_ == std::this_thread::get_id()) {
        AsyncEventGenerator::AsyncStreamAcceptCallBack(&acceptAsyncCtx_);
    } else {
        uv_async_send(&acceptAsyncCtx_);
    }
}

void AsyncEventGenerator::start() {
    threadId_ = getCurrentThreadId();
    uv_run(generatorCoreLoop_, UV_RUN_DEFAULT);
    //停止所有的事件
    Utils::CloseUvLoop(generatorCoreLoop_);
}

Future<bool> AsyncEventGenerator::startAsync() {
    Promise<bool> p;
    Future<bool> f = p.getFuture();
    std::thread th(std::bind([this, p = std::move(p)]() mutable {
        p.setValue(YES);
        start();
    }));
    th.detach();
    return f;
}

void AsyncEventGenerator::stop() {
    stopAsync().get();
}

Future<bool> AsyncEventGenerator::stopAsync() {
    CHECK(!isClosing_) << "AsyncEventGenerator 正在关闭中,重复调用了停止方法,严重错误!";
    std::lock_guard<std::recursive_mutex> guard(internalLock_);
    if (isClosing_) {
        goto futureOut;
    }
    isClosing_ = YES;
    if (getCurrentThreadId() == threadId_) {
        AsyncEventGenerator::GeneratorAsyncStopCb(&closeAsyncCtx_);
    } else {
        uv_async_send(&closeAsyncCtx_);
    }
    futureOut:
    return stopFuture();
}


Future<bool> AsyncEventGenerator::stopFuture() {
    return closePromise_.getFuture();
}


void AsyncEventGenerator::UninstallHandlesCb(uv_handle_t *hd) {
    AsyncEventGenerator *generator = (AsyncEventGenerator *) hd->data;
    if (--generator->openInHandle_ == 0) {
        generator->closePromise_.setValue(YES);
        uv_stop(generator->generatorCoreLoop_);
        delete generator;
    }
}

void AsyncEventGenerator::GeneratorAsyncStopCb(uv_async_t *async) {
    AsyncEventGenerator *s = (AsyncEventGenerator *) async->data;
    SYS_FAULT(&s->closeAsyncCtx_ == async) << "非法调用 GeneratorAsyncStopCb";
    uv_close((uv_handle_t *) &s->acceptAsyncCtx_, &AsyncEventGenerator::UninstallHandlesCb);
    uv_close((uv_handle_t *) &s->closeAsyncCtx_, &AsyncEventGenerator::UninstallHandlesCb);
}

void AsyncEventGenerator::AsyncStreamAcceptCallBack(uv_async_t *async) {
    AsyncEventGenerator *server = (AsyncEventGenerator *) async->data;
    std::lock_guard<std::recursive_mutex> guard(server->internalLock_);
    for (; !server->acceptQueue_.empty();) {
        AbstractAsyncStream *stream = server->acceptQueue_.front();
        stream->attach(server);
        server->acceptQueue_.pop_front();
    }
}

ThreadId AsyncEventGenerator::getEventGeneratorThreadId() {
    return threadId_;
}

uv_loop_t *AsyncEventGenerator::getEventGeneratorCoreLoop() {
    return generatorCoreLoop_;
}

RawDataType *AsyncEventGenerator::getReadBuffer() {
    return readBuffer_;
}

size_t AsyncEventGenerator::getReadBufferSize() {
    return readBufferSize_;
}
